#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <amqp.h>
#include <amqp_tcp_socket.h>

void die_on_error(int x, char const *context) {
    if (x < 0) {
        fprintf(stderr, "%s: %s\n", context, amqp_error_string2(x));
        exit(1);
    }
}

void die_on_amqp_error(amqp_rpc_reply_t x, char const *context) {
    switch (x.reply_type) {
    case AMQP_RESPONSE_NORMAL:
        return;

    case AMQP_RESPONSE_NONE:
        fprintf(stderr, "%s: missing RPC reply type!\n", context);
        break;

    case AMQP_RESPONSE_LIBRARY_EXCEPTION:
        fprintf(stderr, "%s: %s\n", context, amqp_error_string2(x.library_error));
        break;

    case AMQP_RESPONSE_SERVER_EXCEPTION:
        switch (x.reply.id) {
        case AMQP_CONNECTION_CLOSE_METHOD: {
            amqp_connection_close_t *m = (amqp_connection_close_t *)x.reply.decoded;
            fprintf(stderr, "%s: server connection error %uh, message: %.*s\n",
                    context,
                    m->reply_code,
                    (int)m->reply_text.len, (char *)m->reply_text.bytes);
            break;
        }
        case AMQP_CHANNEL_CLOSE_METHOD: {
            amqp_channel_close_t *m = (amqp_channel_close_t *)x.reply.decoded;
            fprintf(stderr, "%s: server channel error %uh, message: %.*s\n",
                    context,
                    m->reply_code,
                    (int)m->reply_text.len, (char *)m->reply_text.bytes);
            break;
        }
        default:
            fprintf(stderr, "%s: unknown server error, method id 0x%08X\n", context, x.reply.id);
            break;
        }
        break;
    }

    exit(1);
}

int main() {
    // 创建连接状态对象
    amqp_connection_state_t conn = amqp_new_connection();
    
    // 创建TCP套接字
    amqp_socket_t *socket = amqp_tcp_socket_new(conn);
    if (!socket) {
        fprintf(stderr, "创建套接字失败\n");
        return 1;
    }
    
    // 连接到RabbitMQ服务器
    int status = amqp_socket_open(socket, "localhost", 5672);
    if (status) {
        fprintf(stderr, "连接RabbitMQ服务器失败\n");
        return 1;
    }
    
    // 登录到服务器
    amqp_rpc_reply_t reply = amqp_login(conn, "/", 0, 131072, 0, 
                                       AMQP_SASL_METHOD_PLAIN, 
                                       "guest", "guest");
    die_on_amqp_error(reply, "登录失败");
    
    // 打开通道
    const int channel = 1;
    amqp_channel_open(conn, channel);
    reply = amqp_get_rpc_reply(conn);
    die_on_amqp_error(reply, "打开通道失败");
    
    // 声明队列
    amqp_queue_declare_ok_t *r = amqp_queue_declare(
        conn,
        channel,
        amqp_cstring_bytes("hello"),  // 队列名称
        0,  // passive
        0,  // durable
        0,  // exclusive
        0,  // auto_delete
        amqp_empty_table  // arguments
    );
    
    reply = amqp_get_rpc_reply(conn);
    die_on_amqp_error(reply, "声明队列失败");
    
    // 设置QoS，每次只处理一条消息
    amqp_basic_qos(conn, channel, 0, 1, 0);
    reply = amqp_get_rpc_reply(conn);
    die_on_amqp_error(reply, "设置QoS失败");
    
    // 在指定队列上注册消费者
    amqp_basic_consume(conn, channel, 
                       amqp_cstring_bytes("hello"),  // 队列名称
                       amqp_empty_bytes,             // 消费者标签
                       0,                            // no_local
                       1,                            // no_ack (0=需要确认, 1=自动确认)
                       0,                            // exclusive
                       amqp_empty_table);            // 参数
    
    reply = amqp_get_rpc_reply(conn);
    die_on_amqp_error(reply, "开始消费失败");
    
    printf(" [*] Waiting for messages. To exit press CTRL+C\n");
    
    // 消费消息循环
    while (1) {
        amqp_rpc_reply_t res;
        amqp_envelope_t envelope;
        
        // 获取消息
        res = amqp_consume_message(conn, &envelope, NULL, 0);
        
        if (AMQP_RESPONSE_NORMAL != res.reply_type) {
            break;
        }
        
        // 处理消息
        printf(" [x] Received %.*s\n", 
               (int)envelope.message.body.len, 
               (char *)envelope.message.body.bytes);
        
        // 手动确认消息 (如果no_ack=0)
        // amqp_basic_ack(conn, channel, envelope.delivery_tag, 0);
        
        // 清理消息资源
        amqp_destroy_envelope(&envelope);
    }
    
    // 关闭连接
    amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS);
    amqp_connection_close(conn, AMQP_REPLY_SUCCESS);
    amqp_destroy_connection(conn);
    
    return 0;
}